from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
from pyspark.sql.functions import concat
from pyspark.sql.types import IntegerType,StructType, StructField, StringType,BooleanType,ArrayType,FloatType
from pyspark.sql.functions import *
spark =SparkSession.builder.\
    appName('restaurant').\
    master('local').\
    getOrCreate()

# business_schema = StructType().\
#     add('business_id',StringType(),True).\
#     add('name',StringType(),True).\
#     add('address',StringType(),True).\
#     add('city',StringType(),True). \
#     add('postal_code', StringType(), True). \
#     add('latitude', FloatType(), True). \
#     add('longitude', FloatType(), True). \
# add('stars', IntegerType(), True). \
# add('review_count', IntegerType(), True). \
# add('is_open', IntegerType(), True). \
# add('attributes', StructType(), True). \
# add('categories', ArrayType(StringType), True). \
# add('hours', StructType(), True)




#df.show()

##################################################################
# df =spark.read.\
#     option('header',True).\
#     json('business.json')

# 找出美国最常见商户（前20）
# 找出美国商户最多的前10个城市
# 找出美国商户最多的前5个州
# 找出美国最常见商户，并显示平均评分（前20）
# 统计评分最高的城市（前10）
# df.groupBy('city').\
#     agg(
#     avg('stars').alias('avg_stars')
#     ).\
#     orderBy('avg_stars',ascending=False).\
#     limit(10).\
#     show()
# 统计category的数量
# 统计最多的category及数量（前10）
# df.printSchema()
# 收获五星评论最多的商户（前20）


#######################################################################3
# def get_Category(data):
#     data = str(data)
#     if "Chinese" in data:
#         return "Chinese"
#     elif "American" in data:
#         return "American"
#     elif "Mexican" in data:
#         return "Mexican"
#     else:
#         return "others"
#
# udf_get_category = udf(get_Category,StringType())
#
# df = df.withColumn('cuisine',udf_get_category(df['categories']))

# 统计不同类型（中国菜、美式、墨西哥）的餐厅类型及数量
#
# df.show()
# df.printSchema()

# df.groupBy('cuisine').\
#     agg(count('*')).show()

# 统计不同类型（中国菜、美式、墨西哥）的餐厅的评论数量
# review_df = spark.read.\
#     option('header',True).\
#     json('yelp_academic_dataset_review.sample.json')
# df = df.select('cuisine','business_id')
# df.join(review_df,df['business_id'] == review_df['business_id'])\
#     .groupBy('cuisine')\
#     .count()\
#     .alias('comment_count')\
#     .show()

# 统计不同类型（中国菜、美式、墨西哥）的餐厅的评分分布

#################################3############################################

#                       3

review_df = spark.read.\
    option('header',True).\
    json('yelp_academic_dataset_review.sample.json')
review_df = review_df.withColumn('date',to_timestamp(review_df['date'],'yyyy-MM-dd HH:mm:ss'))\
    .withColumn('year',year(col('date')))
# review_df.printSchema()
# review_df.show()
# 统计每年的评论数
# review_df.groupBy('year')\
#     .count()\
#     .show()
# 统计评分周（周一~周天）次数统计
# review_df = review_df.withColumn('week',dayofweek(col('date')))
# review_df.groupBy('week')\
#     .count()\
#     .show()
# 统计评分的分布情况
# review_df.groupBy('stars')\
#     .count()\
#     .show()
# 统计有用（helpful）、有趣（funny）及酷（cool）的评论及数量

# 每年全部评论用户排行榜

#
# review_df.select('date')\
#     .show()
####添加一行





